package major;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.clients.producer.internals.ProducerInterceptors;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.ArrayList;
import java.util.Map;
import java.util.Properties;
import java.util.Scanner;
import java.util.concurrent.atomic.AtomicInteger;

public class Sender {
    public static void main(String[] args) {
        String topic = "test-second";
        Properties configs = new Properties();
        configs.setProperty("bootstrap.servers","192.168.0.7:9092");
        configs.setProperty(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,"major.CountInter");
        StringSerializer serializer = new StringSerializer();
        KafkaProducer<String, String> producer = new KafkaProducer<>(configs,serializer,serializer);

        producer.send(new ProducerRecord<>(topic,"goodnice-kafka"));
        new Thread(()->{
            Scanner scanner = new Scanner(System.in);
            while (true){
                String s = scanner.nextLine();
                if (s.equals("quit")){
                    break;
                }
                producer.send(new ProducerRecord<>(topic,s));
                producer.flush();
            }
            producer.close();
        }).start();
        for (int i = 200; i <250 ; i++) {
            producer.send(new ProducerRecord<>(topic, String.format("message%04d", i)));
        }
        producer.flush();
    }

}